package org.mxp.demo5;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import org.mxp.utils.RabbitMQUtils;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage {
    public static void main(String[] args) throws Exception {
        String yibu = ConfirmMessage.yibu();
        System.out.println(yibu);
    }

    // 异步发布确认
    public static String yibu() throws Exception {
        // 获取信道
        Channel channel = RabbitMQUtils.getChannel();

        // 生成一个持久化队列
        boolean durable = true;
        channel.queueDeclare("demo5_3", durable, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 记录消息，用于处理异常消息，线程安全
        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();
        /**
         * 消息确认回调的函数
         *  参数1：消息的标记
         *  参数2：是否为批量确认，来源于RabbitMQ，无法设置
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 是否是批量，受RabbitMQ自身的影响，无法手动设置
            if (multiple) {
                // 删除掉已经确认的消息 剩下的就是未确认的消息
                // ConcurrentSkipListMap的headMap会根据传入的值返回小于他的所有key对应的集合映射，并且操作返回的集合会影响原始集合。deliveryTag是当前消息的唯一ID
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息:" + deliveryTag);
        };
        /**
         * 消息确认失败回调函数
         *  参数1：消息的标记
         *  参数2：是否为批量确认，来源于RabbitMQ，无法设置
         */
        // 消息确认失败回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            // 打印一下未确认的消息都有哪些
            String message = outstandingConfirms.remove(deliveryTag);
            // 处理异常消息
            System.out.println("未确认的消息是：" + message+ " --- 未确认的消息tag：" + deliveryTag);
        };
        /**
         * 准备消息的监听器，监听那些消息成功了，哪些消息失败了
         *  参数1：监听哪些消息成功了
         *  参数2：监听哪些消息失败了
         */
        channel.addConfirmListener(ackCallback,nackCallback);
        // 开始时间
        long begin = System.currentTimeMillis();
        // 发送10条消息
        for (int i = 0; i < 1000; i++) {
            // 发送持久化消息
            channel.basicPublish("", "demo5_3", MessageProperties.PERSISTENT_TEXT_PLAIN, (i + "").getBytes());
            // 此处记录下所有要发送的消息 消息的总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(), (i + ""));
        }
        // 结束时间
        long end = System.currentTimeMillis();
        return "消息发送完毕，耗时：" + (end - begin) + "ms";
    }
}
